package com.shujia.flink.sink

import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo3SinkFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //使用自定义sink
    linesDS.addSink(new SinkFunction[String] {
      /**
       * invoke 每一条数据都会执行一次
       *
       * @param value   ： 一条数据
       * @param context ： 上下文对象
       */
      override def invoke(value: String, context: SinkFunction.Context): Unit = {
        println(value)
      }
    })


    env.execute()
  }

}
